package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.common.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.hadoop.mapreduce.ID;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private final OutputTag<JSONObject> dimTag;
    private final MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection conn;

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        // 建立连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override //处理主流数据
    public void processElement(JSONObject jsonObject, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
      // 获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
//        获取表名
        String tableName = jsonObject.getString("table");
//        获取操作类型
        String type = jsonObject.getString("type");
//        注意：如果采用maxwell采集历史数据,那么操作类型是bootstrap-insert，需要改为insert
        if("bootstrap-insert".equals(type)){
            type="insert";
            jsonObject.put("type",type);
        }
//        拼接查询的key
        String key= tableName + ":" + type;
//        根据key去状态中查询对应的配置信息
        TableProcess tableProcess = broadcastState.get(key);
        if (tableProcess != null){
            String sinkTable = tableProcess.getSinkTable();
            jsonObject.put("sink_table",sinkTable);
//            根据配置表的配置 对字段进行过滤
            JSONObject datajsonObj = jsonObject.getJSONObject("data");
            String sinkColumns = tableProcess.getSinkColumns();
            if (sinkColumns != null && sinkColumns.length() >0){
                filterColnmu(datajsonObj,sinkColumns);
            }
            // 找到配置信息了 进行分流
            if (tableProcess.getSinkType().equals(TableProcess.SINK_TYPE_HBASE)){
                //维度数据 -- 发送到维度侧输出流中
                ctx.output(dimTag,jsonObject);
            } else if (tableProcess.getSinkType().equals(TableProcess.SINK_TYPE_KAFKA))
            {
                //事实数据 发送到主流
                out.collect(jsonObject);
                System.out.println("主流");
            }
        } else {
            // 没有找到配置信息
            System.out.println("No this Key in Table Process: " + key);
        }

    }
// 对字段进行过滤
    private void filterColnmu(JSONObject dataJsonObj ,String sinkColumns){
        //对sinkColumns进行分割，得到的是保留的字段和名称属性
        String[] fileds = sinkColumns.split(",");
        List<String> filedList = Arrays.asList(fileds);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(  entry-> !filedList.contains(entry.getKey()));
    }


    @Override //处理广播流数据()
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //将读取到FlinkCDC采集到的信息  由jsonStr->jsonObj
        // {"database":"gmall1116","table":"comment_info","type":"insert","data":{}}
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取data数据  也就是每一条配置信息
        String dataJsonStr = jsonObj.getString("data");
        //将读取到的配置信息转换为TableProcess对象
        TableProcess tableProcess = JSON.parseObject(dataJsonStr, TableProcess.class);

        //获取源表表名
        String sourceTable = tableProcess.getSourceTable();
        //获取操作类型
        String operateType = tableProcess.getOperateType();
        //输出类型      hbase|kafka
        String sinkType = tableProcess.getSinkType();
        //输出目的地表名或者主题名
        String sinkTable = tableProcess.getSinkTable();
        //输出字段
        String sinkColumns = tableProcess.getSinkColumns();
        //表的主键
        String sinkPk = tableProcess.getSinkPk();
        //建表扩展字段
        String sinkExtend = tableProcess.getSinkExtend();
        //拼接key
        String key = sourceTable + ":" + operateType;

        //如果是维度类型配置
        if (sinkType.equals(TableProcess.SINK_TYPE_HBASE) && "insert".equals(operateType)) {
            //通过Phoenix创建表
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }

        //获取广播状态
        BroadcastState<String, TableProcess> tableProcessState = ctx.getBroadcastState(mapStateDescriptor);
        tableProcessState.put(key, tableProcess);
    }

    //创建维度表
    private void checkTable(String tableName, String fieldStr, String pk, String ext) {
        if (ext == null) {
            ext = "";
        }
        if (pk == null) {
            pk = "id";
        }
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.HBASE_SCHEMA + "." + tableName + "(");
        String[] fieldArr = fieldStr.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String field = fieldArr[i];
            if (pk.equals(field)) {
                createSql.append(field + " varchar primary key ");
            } else {
                createSql.append("info." + field + " varchar ");
            }
            if (i < fieldArr.length - 1) {
                createSql.append(",");
            }

        }
        createSql.append(")" + ext);

        System.out.println("Phoenix建表语句:" + createSql);
        //创建表
        PreparedStatement ps = null;
        try {
            //创建数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行SQL语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //释放资源
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
